package voldemort.store.cachestore.volderimpl;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import voldemort.VoldemortException;
import voldemort.store.NoSuchCapabilityException;
import voldemort.store.StorageEngine;
import voldemort.store.StoreCapabilityType;
import voldemort.store.cachestore.BlockSize;
import voldemort.store.cachestore.Key;
import voldemort.store.cachestore.Value;
import voldemort.store.cachestore.impl.CacheStore;
import voldemort.utils.ClosableIterator;
import voldemort.utils.Pair;
import voldemort.versioning.ClockEntry;
import voldemort.versioning.VectorClock;
import voldemort.versioning.Version;
import voldemort.versioning.Versioned;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * Created by IntelliJ IDEA.
 * User: thsieh
 * Date: May 16, 2010
 * Time: 1:36:44 PM
 * To change this template use File | Settings | File Templates.
 */
public class VoldeStore<K, V> implements StorageEngine<K, V> {
    
    private static Log logger = LogFactory.getLog(VoldeStore.class);
//    private ConcurrentMap<Key, CacheBlock> map;
//    // read
//    private final ReentrantLock lock = new ReentrantLock();
    //interface which encapuse the logic of determine storage block size
    private BlockSize blockSize;
    // Channel store
//    private List<ChannelStore> list;
    //storage path at constructor time
//    private String path;
    // current index, which intented to support compaction of deletion
    private int curIndex ;
    // for voldermort interface
    private String storeName;
    // node id of storage
    private short nodeId;
    //public final static String FileNamePrefix = "cachestore";
    private CacheStore cacheStore;

    public VoldeStore(String storeName, BlockSize blockSize, String path, short nodeId) {
        this.storeName = storeName;
        this.blockSize = blockSize;
//        this.path = path;
        this.nodeId = nodeId;
//        this.map = new ConcurrentHashMap<Key, CacheBlock>(119);
//        this.list = new CopyOnWriteArrayList<ChannelStore>();
        this.curIndex = 0;
        //this.list.add( open(getFileName(storeName, curIndex), map) );
        this.cacheStore = new CacheStore( path, blockSize, curIndex, storeName );
    }




    public ClosableIterator<Pair<K, Versioned<V>>> entries() {
        logger.warn("No support ClosableIterator");
        return null;
    }

    public ClosableIterator<K> keys() {
        logger.warn("No support ClosableIterator");
        return null;
    }

    public void truncate() {
        logger.warn("No support for truncate");
    }

    private Versioned get(Key k) {
        Value value = cacheStore.get( k);
        if ( value == null ) return null;
        else return  makeVersioned4Block( value) ;
    }

    private Versioned makeVersioned4Block(Value value) {
        List<ClockEntry> list = new ArrayList<ClockEntry>(1);
        list.add(new ClockEntry(value.getNode(), value.getVersion()) ) ;
        VectorClock clock = new VectorClock(list, System.currentTimeMillis());
        return  new Versioned(value.getData(), clock);
    }

    /**
     *
     * @param key - defined by vlodermort interface
     * @return List<Versioned<V>> - Value defined by voldermort
     * @throws VoldemortException
     */
    public List<Versioned<V>> get(K key) throws VoldemortException {
        Key k = Key.createKey( key);
        Versioned v = get( k);
        if ( v == null ) return new ArrayList<Versioned<V>>(0);
        else {
            List<Versioned<V>> list =  new ArrayList<Versioned<V>>(1);
            list.add(v);
            return list;
        }
    }

    /**
     * this is the real interface to channel store
     * @param key - Key class defined by storage
     * @param value - Value interface defined by storage
     */

    private void put(Key key, Value<byte[]> value) {
        cacheStore.put(key, value);

    }

    /**
     *
     * @param keys The keys to check for.
     * @return
     * @throws VoldemortException
     */
    public Map<K, List<Versioned<V>>> getAll(Iterable<K> keys) throws VoldemortException {
        logger.warn("not support getAll");
        return null;
    }

    /**
     *
     * @param key The key to use
     * @param value The value to store and its version.
     * @throws VoldemortException
     */
    public void put(K key, Versioned<V> value) throws VoldemortException {
        Key k = Key.createKey( key);
        long version =  getVersion( (VectorClock) value.getVersion());
        short node = getNode((VectorClock) value.getVersion());
        logger.info("put version "+value.getVersion().toString()+ " node "+ node+" v "+version );
        if ( ! (value.getValue() instanceof byte[])) throw new VoldemortException("value "+ value.getValue().getClass().getName()+
                " must be byte[]");
        //using nodeId in the versioned
        Value v = new BlockValue( value.getValue(), version, node);
        put(k, v);
    }

    private short getNode(VectorClock clock) {
        if ( clock.getEntries().size() == 0 ) return (short) nodeId;
        else {
            // get the last one in the entry
            return clock.getEntries().get(clock.getEntries().size()-1).getNodeId() ;
        }
    }

    private long getVersion(VectorClock clock) {
        long toReturn = clock.getMaxVersion();
        if ( clock.getEntries().size() > 1 )
            toReturn ++;
        return ( toReturn < 1 ? 1 : toReturn);
    }

    /**
     *
     * @param key The key to delete
     * @param version The current value of the key
     * @return
     * @throws VoldemortException
     */
    public boolean delete(K key, Version version) throws VoldemortException {
        Key k = Key.createKey( key);
        return cacheStore.remove( k);
    }

    public String getName() {
        return storeName;
    }

    public void close() throws VoldemortException {
        cacheStore.close();
    }

    public Object getCapability(StoreCapabilityType capability) {
        throw new NoSuchCapabilityException(capability, getName());
    }

    /**
     *
     * @param key
     * @return
     */
    public List<Version> getVersions(K key) {
        List<Version> list = new ArrayList<Version>();
        Key k = Key.createKey( key);
        Versioned ver = get(k);
        // first version == 0 , when it is null
        if ( ver == null) list.add( new VectorClock());
        else list.add( ver.getVersion() );
        return list;
    }

    /**
     * Class that implement channel storage Value interface to bridge Voldermort versioned
     */
    class BlockValue implements Value<V> {
        V data;
        long version;
        short node;

        BlockValue(V data, long version, short node) {
            this.data = data;
            this.version = version;
            this.node = node;
        }

        public V getData() {
            return data;
        }

        public void setData(V data) {
            this.data = data;
        }

        public long getVersion() {
            return version;
        }

        public void setVersion(long version) {
            this.version = version;
        }

        public short getNode() {
            return node;
        }

        public void setNode(short node) {
            this.node = node;
        }
    }
}
